package cn.ipanel.bigdata.job.ods

import cn.ipanel.bigdata.boot.date.{Date, Day, Hour}
import cn.ipanel.bigdata.boot.logger.Logger
import cn.ipanel.bigdata.dw.ods
import cn.ipanel.bigdata.dw.ods.{T_REPORT_BY_HOUR => T}
import cn.ipanel.bigdata.utils.Util.isEmpty
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.{col, lit}

/**
 * @author lzz
 * @projectName iBigdata
 * @date 2023/08/25 14:08
 * @description: 按小时清洗上报，跟 Report 任务逻辑保持一致
 */
class ReportByHour extends Report {

  override def optTargetTable: ods.AbsOdsHive = T

  final lazy val hour: Hour = if(isEmpty(getParam("hour"))) Date.asHour.prev() else Hour(getParam("hour"))

  override def onPrepare(): Unit = {
    _T.dropByHour(hour)
  }

  override def source(): DataFrame = {
    Logger.I("hour: " + hour.toFullHour)
    ods.T_HDFS_LOG.readToTableByHour(hour)
      .filter(col(T.F_TIMESTAMP).between(lit(hour.st), lit(hour.et)))
  }

  // 不能省略，必须重写否则父方法调用 data.filter(col(T.F_DATE) === lit(_DAY.toDate)) 否则会用天过滤时，小时数据会丢
  override def intent(data: DataFrame, day: Day): DataFrame = {
    data.withColumn(T.F_DATE, lit(hour.toFullHour.toInt))
  }

//  不必重写 join，使用 Report 的即可
//  override def join(data: DataFrame, day: Day): DataFrame = {
//
//  }

  override def onFruited(): Boolean = T.findByHour(hour).limit(1).count() > 0

}
